package com.gsm.state;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class d7_ListStateExample {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> input = env.fromElements(
            Tuple2.of("a", 1),
            Tuple2.of("b", 2),
            Tuple2.of("a", 3)
        );

        DataStream<Tuple2<String, Integer>> result = input.keyBy(0).map(new RichMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            private transient ListState<Integer> listState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ListStateDescriptor<Integer> descriptor = new ListStateDescriptor<>(
                    "listState",
                    Integer.class
                );

                listState = getRuntimeContext().getListState(descriptor);
            }

            @Override
            public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                // Add the current value to the list state
                listState.add(value.f1);

                // Get all values from the list state
                Iterable<Integer> allValues = listState.get();
                int sum = 0;
                for (Integer v : allValues) {
                    sum += v;
                }

                return Tuple2.of(value.f0, sum);
            }
        });

        result.print();
        env.execute("ListState Example");
    }
}